package com.surfilter.massdata.spark.task.statistics;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.JdbcUtil;

/**
 *  月表统计逻辑：
	涉及的表有：SA5401、SA6401、SA7400、SAN013、SAN043、SAN063、SA3400、SAN053、SAN083、SAN073
	统计访问量的表有：SA7400、SAN013、SAN043、SAN063、SAN053、SAN083、SAN073
	统计个数的表有：SA5401、SA6401、SA3400
	月表中，每天按照月纬度（天表month字段）进行统计。月表中每天插入本月最新的一条数据，删除本月昨日统计数据。月表中count_value字段存储本月天表中按照month字段纬度统计的count_value字段总和。
	备注：对于count_value字段统计的是访问量，月表中的count_value字段直接累加；对于count_value字段统计的是个数，例如topdamin个数、cip个数等，月表中的count_value字段要先排重再累加
 *
 */
public class MonthStatisticsTask {
	
	private static final Log log = LogFactory.getLog(MonthStatisticsTask.class);
	
	public static void monthStatistics(Date date,Connection conn){
		try{
			log.info("===========月表统计开始============");
			
			//SA5401 行业发展情况统计分析-网站数量变化统计-->对应天表SA5601
//	        String table = "SA5401";
//	        String dayTable = "SA5601";
//	        String buss_type = "WEBSITE_COUNT_SUM";
//	        String buss_value = "1";
//	        int sta_range = 0;
//	        createSeqIfNotExists(table,conn);
//	        insertMonthTable(date,conn,table,dayTable,buss_type,buss_value,sta_range);
	        
	        //SA6401行业发展情况统计分析-域名数量变化统计 -->对应天表SA6601
	       // createSeqIfNotExists("SA6401",conn);
	       // insertMonthTable(date,conn,"SA6401","SA6601","WEBSITE_COUNT_SUM","",0);
	        
	        //SA7400 网民行为分析-网站访问量排名-本省接入-本省主体-分类网站访问量排名-->对应天表SA7601
	        createSeqIfNotExists("SA7400",conn);
	        insertMonthTableCount(date,conn,"SA7400","SA7601","WEBSITE_RANK_BUSS","type",0);
	        insertMonthTableCount(date,conn,"SA7400","SA7601","WEBSITE_RANK_SUM","",0);
	        insertMonthTableCount(date,conn,"SA7400","SA7601","WEBSITE_RANK_SUM","0",2);
	        
	        //SA7400 网民行为分析-网站访问量排名-全球 -->对应天表SA7602
	        insertMonthTableCount(date,conn,"SA7400","SA7602","WEBSITE_RANK_SUM","",1);
	        
	        //SAN013 访问全球网站情况-->对应天表SAN011
	        createSeqIfNotExists("SAN013",conn);
	        insertMonthTable(date,conn,"SAN013","SAN011","WEBSITE_COUNT_SUM","1",0);
	        insertMonthTable(date,conn,"SAN013","SAN011","WEBSITE_COUNT_SUM","2",0);
	        //访问省内网站情况 
	        insertMonthTableCount1(date,conn,"SAN013","SAN011","WEBSITE_COUNT_AREA","",0);
	        
	        //SAN043 运营商网站访问量-->对应天表SAN041
	        createSeqIfNotExists("SAN043",conn);
	        insertMonthTableCount1(date,conn,"SAN043","SAN041","WEBSITE_COUNT_OPERATORS","",0);
	        //行业分类网站访问量
	        insertMonthTableCount1(date,conn,"SAN043","SAN041","WEBSITE_COUNT_BUSS","",0);
	        
	        //SAN063 网站后缀访问量-->对应天表SAN061
	        createSeqIfNotExists("SAN063",conn);
	        insertMonthTableCount1(date,conn,"SAN063","SAN061","WEBSITE_COUNT_SUFFIX","",0);
	        
	        //SA3400 我省活跃用户IP统计-->对应天表SA3601
	        //createSeqIfNotExists("SA3400",conn);
	        //insertMonthTableCount1(date,conn,"SA3400","SA3601","WEBSITE_COUNT_SUM","PROVINCE_COUNT",0);
	        //我省活跃用户IP数量分布统计-运营商用户IP数量分布
	        //insertMonthTableCount1(date,conn,"SA3400","SA3601","WEBSITE_COUNT_OPERATORS","",0);
	        //我省活跃用户IP数量分布统计-地市用户IP数量
	       // insertMonthTableCount1(date,conn,"SA3400","SA3601","WEBSITE_COUNT_AREA","",0);
	        //我省活跃用户IP数量分布统计-用户类型用户IP数量
	       // insertMonthTableCount1(date,conn,"SA3400","SA3601","WEBSITE_COUNT_USE_UNIT_TYPE","",0);
	        
	        //SAN053 活跃IP排名 -->对应天表SAN051
	        createSeqIfNotExists("SAN053",conn);
	        insertMonthTableCount2(date,conn,"SAN053","SAN051","WEBSITE_COUNT_OPERATORS","",0);
	        //地市用户IP访问量排名
	        insertMonthTableCount2(date,conn,"SAN053","SAN051","WEBSITE_COUNT_AREA","",0);
	       
	        //SAN083 地市网民访问量-->对应天表SAN081
	        createSeqIfNotExists("SAN083",conn);
	        insertMonthTableCount1(date,conn,"SAN083","SAN081","COUNT_AREA_SUM","",0);
	        //运营商网民访问量
	        insertMonthTableCount1(date,conn,"SAN083","SAN081","COUNT_OPERATORS_SUM","",0);
	        
	        //SAN073 网民访问量分类统计-->对应天表SAN071
	        createSeqIfNotExists("SAN073",conn);
	        insertMonthTableCount1(date,conn,"SAN073","SAN071","COUNT_IP_TYPE","",0);
	        
	        log.info("===========月表统计结束===========");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}

	private static void insertMonthTableCount2(Date date,Connection conn,String table,String dayTable,String buss_type,String buss_value,int sta_range) throws SQLException {
		PreparedStatement ps = null;
		PreparedStatement stmt = null;
		ResultSet rs = null;
		try{
			int year = DateUtil.getCurrentYear(date);
			int month = DateUtil.getCurrentMonth(date);
			StringBuffer queryCond = new StringBuffer();
			queryCond.append(" where year = "+year);
			queryCond.append("   and month = "+month);
			queryCond.append("   and buss_type = '"+buss_type+"'");
			queryCond.append("   and sta_range = "+sta_range);
			if(StringUtils.isNotBlank(buss_value)){
				queryCond.append("   and buss_value = '"+buss_value+"'");
			}
			
			StringBuffer sqlBuffer = new StringBuffer();
			sqlBuffer.append("select buss_value, client_ip, sum(count_value) count_value");
			sqlBuffer.append("  from "+dayTable);
			sqlBuffer.append(queryCond);
			sqlBuffer.append(" group by buss_value, client_ip");
			sqlBuffer.append(" order by buss_value,count_value desc");
			
			ps = conn.prepareStatement(sqlBuffer.toString());
	        rs = ps.executeQuery();
	        
	        List<Map<String,Object>> list = resultSetToList(rs);
	        
	        log.info("天表【"+dayTable+"】-->月表【"+table+"】");
	        log.info("月统计集合数量："+list.size());
	        
	        String sql = "";
	        boolean flag = false;
	        if(list.size() > 0){
	        	//删除当前月昨日统计数据
	        	sqlBuffer = new StringBuffer();
	        	sqlBuffer.append("delete from "+table+"");
	        	sqlBuffer.append(queryCond);
	        	flag = execSql(conn,sqlBuffer.toString());
	        	if(flag){
	        		//插入当前月统计最新数据                                           
	        		sql = "insert into "+table+"(id,year,half_year,quarter,month,week,day,hour,buss_type,buss_value,count_value,sta_range,client_ip,rank) values("+table+"_SE.nextval,?,?,?,?,?,?,?,?,?,?,?,?,?)";
	        		conn.setAutoCommit(false);
	        		stmt = conn.prepareStatement(sql);
	        		int n = 0;
	        		String bussValue = CommonUtils.valueOf(list.get(0).get("BUSS_VALUE"));
	        		int rank = 1;//排名
	        		for(Map<String,Object> map : list){
	        			//不同的纬度进行排名
	        			String bussValue1 = CommonUtils.valueOf(map.get("BUSS_VALUE"));
	        			if(!StringUtils.equals(bussValue1,bussValue)){
	        				bussValue = bussValue1;
	        				rank = 1;
	        			}
	        			stmt.setObject(1,year);
	        			stmt.setObject(2,DateUtil.getHalfYear(date));
	        			stmt.setObject(3,DateUtil.getQuarter(date));
	        			stmt.setObject(4,month);
	        			stmt.setObject(5,DateUtil.getCurrentWeek(date));
	        			stmt.setObject(6,DateUtil.getCurrentDay(date));
	        			stmt.setObject(7,0);
	        			stmt.setObject(8,buss_type);
	        			stmt.setObject(9,map.get("BUSS_VALUE"));
	        			long count_value=0l;
	        			if(map.get("COUNT_VALUE")==null||StringUtils.isBlank(map.get("COUNT_VALUE").toString())||map.get("COUNT_VALUE").toString().equalsIgnoreCase("null")){
	        				count_value=0l;
	        			}else{
	        				count_value=Long.valueOf(map.get("COUNT_VALUE").toString());
	        			}
	        			stmt.setObject(10,count_value);
	        			stmt.setObject(11,sta_range);
	        			stmt.setObject(12,map.get("CLIENT_IP"));
	        			stmt.setObject(13,rank);
	        			rank++;
	        			stmt.addBatch();
	        			if (++n % 5000 == 0) {
	        				stmt.executeBatch();
	        				conn.commit();
						}
	        		 }
	        		stmt.executeBatch();
					conn.commit();
	        	 }
	         }
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(rs);
			JdbcUtil.close(ps);
			JdbcUtil.close(stmt);
		}
	}
	
	private static void insertMonthTableCount(Date date,Connection conn,String table,String dayTable,String buss_type,String busss_value,int sta_range) throws SQLException {
		PreparedStatement ps = null;
		PreparedStatement stmt = null;
		ResultSet rs = null;
		try{
			int year = DateUtil.getCurrentYear(date);
			int month = DateUtil.getCurrentMonth(date);
			StringBuffer queryCond = new StringBuffer();
			queryCond.append(" where year = "+year);
			queryCond.append("   and month = "+month);
			queryCond.append("   and buss_type = '"+buss_type+"'");
			queryCond.append("   and sta_range = "+sta_range);
			if(StringUtils.isBlank(busss_value)){
				queryCond.append("   and buss_value is null");
			}else if(StringUtils.equals(busss_value, "0")){
				queryCond.append("   and buss_value = '0'");
			}
			
			StringBuffer sqlBuffer = new StringBuffer();
			sqlBuffer.append("select buss_value, domain, website_name, icp_num, sum(traffic) traffic");
			sqlBuffer.append("  from "+dayTable);
			sqlBuffer.append(queryCond);
			sqlBuffer.append(" group by buss_value, domain, website_name, icp_num");
			sqlBuffer.append(" order by buss_value,traffic desc");
			
			ps = conn.prepareStatement(sqlBuffer.toString());
	        rs = ps.executeQuery();
	        
	        List<Map<String,Object>> list = resultSetToList(rs);
	        
	        log.info("天表【"+dayTable+"】-->月表【"+table+"】");
	        log.info("月统计集合数量："+list.size());
	        
	        String sql = "";
	        boolean flag = false;
	        if(list.size() > 0){
	        	//删除当前月昨日统计数据
	        	sqlBuffer = new StringBuffer();
	        	sqlBuffer.append("delete from "+table+"");
	        	sqlBuffer.append(queryCond);
	        	flag = execSql(conn,sqlBuffer.toString());
	        	if(flag){
	        		//插入当前月统计最新数据                                    
	        		sql = "insert into "+table+"(id,year,half_year,quarter,month,week,day,buss_type,buss_value,rank,domain,website_name,icp_num,traffic,sta_range) values("+table+"_SE.nextval,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
	        		conn.setAutoCommit(false);
	        		stmt = conn.prepareStatement(sql);
	        		int n = 0;
	        		String bussValue = CommonUtils.valueOf(list.get(0).get("BUSS_VALUE"));
	        		int rank = 1;
	        		for(Map<String,Object> map : list){
	        			//不同的纬度进行排名
	        			String bussValue1 = CommonUtils.valueOf(map.get("BUSS_VALUE"));
	        			if(!StringUtils.equals(bussValue1,bussValue)){
	        				bussValue = bussValue1;
	        				rank = 1;
	        			}
	        			stmt.setObject(1,year);
	        			stmt.setObject(2,DateUtil.getHalfYear(date));
	        			stmt.setObject(3,DateUtil.getQuarter(date));
	        			stmt.setObject(4,month);
	        			stmt.setObject(5,DateUtil.getCurrentWeek(date));
	        			stmt.setObject(6,DateUtil.getCurrentDay(date));
	        			stmt.setObject(7,buss_type);
	        			stmt.setObject(8,map.get("BUSS_VALUE"));
	        			stmt.setObject(9,rank);
	        			stmt.setObject(10,map.get("DOMAIN"));
	        			stmt.setObject(11,map.get("WEBSITE_NAME"));
	        			stmt.setObject(12,map.get("ICP_NUM"));
	        			stmt.setObject(13,map.get("TRAFFIC"));
	        			stmt.setObject(14,sta_range);
	        			rank++;
	        			stmt.addBatch();
	        			if (++n % 5000 == 0) {
	        				stmt.executeBatch();
	        				conn.commit();
						}
	        		 }
	        		stmt.executeBatch();
					conn.commit();
	        	 }
	         }
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(rs);
			JdbcUtil.close(ps);
			JdbcUtil.close(stmt);
		}
	}
	
	private static void insertMonthTableCount1(Date date,Connection conn,String table,String dayTable,String buss_type,String buss_value,int sta_range) throws SQLException {
		PreparedStatement ps = null;
		PreparedStatement stmt = null;
		ResultSet rs = null;
		try{
			int year = DateUtil.getCurrentYear(date);
			int month = DateUtil.getCurrentMonth(date);
			StringBuffer queryCond = new StringBuffer();
			queryCond.append(" where year = "+year);
			queryCond.append("   and month = "+month);
			queryCond.append("   and buss_type = '"+buss_type+"'");
			queryCond.append("   and sta_range = "+sta_range);
			if(StringUtils.isNotBlank(buss_value)){
				queryCond.append("   and buss_value = '"+buss_value+"'");
			}
			
			StringBuffer sqlBuffer = new StringBuffer();
			sqlBuffer.append("select buss_value,sum(count_value) as count_value");
			sqlBuffer.append("  from "+dayTable+"");
			sqlBuffer.append(queryCond);
			sqlBuffer.append("  group by buss_value");
			
			ps = conn.prepareStatement(sqlBuffer.toString());
	        rs = ps.executeQuery();
	        
	        List<Map<String,Object>> list = resultSetToList(rs);
	        
	        log.info("天表【"+dayTable+"】-->月表【"+table+"】");
	        log.info("月统计集合数量："+list.size());
	        
	        String sql = "";
	        boolean flag = false;
	        if(list.size() > 0){
	        	//删除当前月昨日统计数据
	        	sqlBuffer = new StringBuffer();
	        	sqlBuffer.append("delete from "+table+"");
	        	sqlBuffer.append(queryCond);
	        	flag = execSql(conn,sqlBuffer.toString());
	        	if(flag){
	        		
	        		//插入当前月统计最新数据                                           
	        		sql = "insert into "+table+"(id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range) values("+table+"_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
	        		conn.setAutoCommit(false);
	        		stmt = conn.prepareStatement(sql);
	        		int n = 0;
	        		for(Map<String,Object> map : list){
	        			stmt.setObject(1,year);
	        			stmt.setObject(2,DateUtil.getHalfYear(date));
	        			stmt.setObject(3,DateUtil.getQuarter(date));
	        			stmt.setObject(4,month);
	        			stmt.setObject(5,DateUtil.getCurrentWeek(date));
	        			stmt.setObject(6,DateUtil.getCurrentDay(date));
	        			stmt.setObject(7,buss_type);
	        			stmt.setObject(8,map.get("BUSS_VALUE"));
	        			long count_value=0l;
	        			if(map.get("COUNT_VALUE")==null||StringUtils.isBlank(map.get("COUNT_VALUE").toString())||map.get("COUNT_VALUE").toString().equalsIgnoreCase("null")){
	        				count_value=0l;
	        			}else{
	        				count_value=Long.valueOf(map.get("COUNT_VALUE").toString());
	        			}
	        			stmt.setObject(9,count_value);
	        			stmt.setObject(10,sta_range);
	        			 
	        			stmt.addBatch();
	        			if (++n % 5000 == 0) {
	        				stmt.executeBatch();
	        				conn.commit();
						}
	        		 }
	        		stmt.executeBatch();
					conn.commit();
	        	 }
	         }
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(rs);
			JdbcUtil.close(ps);
			JdbcUtil.close(stmt);
		}
	}
	
	private static void insertMonthTable(Date date,Connection conn,String table,String dayTable,String buss_type,String buss_value,int sta_range) throws SQLException {
		PreparedStatement ps = null;
		PreparedStatement stmt = null;
		ResultSet rs = null;
		try{
			int year = DateUtil.getCurrentYear(date);
			int month = DateUtil.getCurrentMonth(date);
			StringBuffer queryCond = new StringBuffer();
			queryCond.append(" where year = "+year);
			queryCond.append("   and month = "+month);
			queryCond.append("   and buss_type = '"+buss_type+"'");
			if(StringUtils.isNotBlank(buss_value)){
				queryCond.append("   and buss_value = '"+buss_value+"'");
			}else{
				queryCond.append("   and buss_value is null");
			}
			queryCond.append("   and sta_range = "+sta_range);
			
			StringBuffer sqlBuffer = new StringBuffer();
			sqlBuffer.append("select sum(count_value) as count_value");
			sqlBuffer.append("  from "+dayTable+"");
			sqlBuffer.append(queryCond);
			
			ps = conn.prepareStatement(sqlBuffer.toString());
	        rs = ps.executeQuery();
	        
	        List<Map<String,Object>> list = resultSetToList(rs);
	        
	        log.info("天表【"+dayTable+"】-->月表【"+table+"】");
	        log.info("月统计集合数量："+list.size());
	        
	        String sql = "";
	        boolean flag = false;
	        if(list.size() > 0){
	        	//删除当前月昨日统计数据
	        	sqlBuffer = new StringBuffer();
	        	sqlBuffer.append("delete from "+table+"");
	        	sqlBuffer.append(queryCond);
	        	flag = execSql(conn,sqlBuffer.toString());
	        	if(flag){
	        		//插入当前月统计最新数据                                    
	        		sql = "insert into "+table+"(id,year,half_year,quarter,month,week,day,buss_type,buss_value,count_value,sta_range) values("+table+"_SE.nextval,?,?,?,?,?,?,?,?,?,?)";
	        		conn.setAutoCommit(false);
	        		stmt = conn.prepareStatement(sql);
	        		int n = 0;
	        		for(Map<String,Object> map : list){
	        			stmt.setObject(1,year);
	        			stmt.setObject(2,DateUtil.getHalfYear(date));
	        			stmt.setObject(3,DateUtil.getQuarter(date));
	        			stmt.setObject(4,month);
	        			stmt.setObject(5,DateUtil.getCurrentWeek(date));
	        			stmt.setObject(6,DateUtil.getCurrentDay(date));
	        			stmt.setObject(7,buss_type);
	        			stmt.setObject(8,buss_value);
	        			long count_value=0l;
	        			if(map.get("COUNT_VALUE")==null||StringUtils.isBlank(map.get("COUNT_VALUE").toString())||map.get("COUNT_VALUE").toString().equalsIgnoreCase("null")){
	        				count_value=0l;
	        			}else{
	        				count_value=Long.valueOf(map.get("COUNT_VALUE").toString());
	        			}
	        			stmt.setObject(9,count_value);
	        			stmt.setObject(10,sta_range);
	        			 
	        			stmt.addBatch();
	        			if (++n % 5000 == 0) {
	        				stmt.executeBatch();
	        				conn.commit();
						}
	        		 }
	        		stmt.executeBatch();
					conn.commit();
	        	 }
	         }
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(rs);
			JdbcUtil.close(ps);
			JdbcUtil.close(stmt);
		}
	}
	
	private static boolean execSql(Connection conn,String sql){
    	boolean flag = true;
    	PreparedStatement ps = null;
    	try{
    		conn.setAutoCommit(false);
    		ps = conn.prepareStatement(sql);
          	ps.executeQuery();
            conn.commit();
          	flag = true;
          }catch(SQLException e){
         	 flag = false;
         	 log.error("jdbc batch commit sql : "+sql+" error" , e);
              try {
                  conn.rollback();
              } catch (SQLException e1) {
                  log.error("jdbc rollback error" , e1);
              }
          }finally{
        	  JdbcUtil.close(ps);
          }
    	return flag ;
    }
	
	public static List<Map<String, Object>> resultSetToList(ResultSet rs) throws SQLException{
		if (rs == null) return new ArrayList<Map<String, Object>>();
		ResultSetMetaData md = rs.getMetaData(); // 得到结果集(rs)的结构信息，比如字段数、字段名等
		int columnCount = md.getColumnCount(); // 返回此 ResultSet 对象中的列数
		List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
		Map<String, Object> rowData = new HashMap<String, Object>();
		while (rs.next()) {
			rowData = new HashMap<String, Object>(columnCount);
			for (int i = 1; i <= columnCount; i++) {
				rowData.put(md.getColumnName(i), rs.getObject(i));
			}
			list.add(rowData);
		}
		return list;
	}
	public static void createSeqIfNotExists(String tableName,Connection conn){
		try{
			String sequenceName = tableName.toUpperCase() + "_SE";
			CommonUtils.getSequenceName(sequenceName,conn);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
	}
}
